package com.ddone.demo;

import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.Text;

import java.util.List;

@Description(name = "is_point_in_aoi", value = "_FUNC_(Array<String> aoi, String point) - returns true if point is inside AOI")
public class IsPointInAOIGenericUDF extends GenericUDF {

    private PrimitiveObjectInspector stringOI;
    private PrimitiveObjectInspector listOI;

    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
        if (arguments.length != 2) {
            throw new UDFArgumentLengthException("The function is_point_in_aoi requires 2 arguments.");
        }

        if (!(arguments[0] instanceof PrimitiveObjectInspector) || !(arguments[1] instanceof PrimitiveObjectInspector)) {
            throw new UDFArgumentTypeException(0, "The first argument must be an array of strings (AOI coordinates), and the second must be a string representing the point.");
        }

        listOI = (PrimitiveObjectInspector) arguments[0];
        stringOI = (PrimitiveObjectInspector) arguments[1];

        return PrimitiveObjectInspectorFactory.writableBooleanObjectInspector;
    }

    @Override
    public Object evaluate(DeferredObject[] arguments) throws HiveException {
        List<Text> aoiCoordinates = (List<Text>) arguments[0].get();
        String point = PrimitiveObjectInspectorUtils.getString(arguments[1].get(), stringOI);

        if (aoiCoordinates == null || point == null) {
            return new BooleanWritable(false);
        }

        String[] pointParts = point.split("_");
        if (pointParts.length != 2) {
            throw new HiveException("The point must be in 'lon_lat' format.");
        }

        double pointLon = Double.parseDouble(pointParts[0]);
        double pointLat = Double.parseDouble(pointParts[1]);

        int n = aoiCoordinates.size();
        boolean inside = false;

        // 解析多边形的坐标点
        double[] lon = new double[n];
        double[] lat = new double[n];
        for (int i = 0; i < n; i++) {
            String[] parts = aoiCoordinates.get(i).toString().split("_");
            lon[i] = Double.parseDouble(parts[0]);
            lat[i] = Double.parseDouble(parts[1]);
        }

        // 射线法判定点是否在多边形内
        for (int i = 0, j = n - 1; i < n; j = i++) {
            if (((lat[i] > pointLat) != (lat[j] > pointLat)) &&
                    (pointLon < (lon[j] - lon[i]) * (pointLat - lat[i]) / (lat[j] - lat[i]) + lon[i])) {
                inside = !inside;
            }
        }

        return new BooleanWritable(inside);
    }

    @Override
    public String getDisplayString(String[] children) {
        return "is_point_in_aoi(" + children[0] + ", " + children[1] + ")";
    }
}
